package de.distMLP.communication;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

import org.apache.hama.bsp.BSPPeer;
import org.apache.hama.bsp.sync.SyncException;

import de.distMLP.messages.HamaMessage;

public abstract class MessageDistributor {

	private final BSPPeer<?, ?, ?, ?, HamaMessage> peer;
	private final Map<Integer, String> peerMap;
	private final Map<String, Integer> peerToIntMap;

	private final Map<Integer, String> localPeerMap;
	private final Map<String, Integer> localPeerToIntMap;

	private final String masterPeer;
	private final boolean isLocalMode;

	public MessageDistributor(final BSPPeer<?, ?, ?, ?, HamaMessage> peer, final String masterPeer) {
		this.peer = peer;
		this.masterPeer = masterPeer;
		this.isLocalMode = this.isLocalŔunningMode();

		this.localPeerMap = new HashMap<Integer, String>();
		this.localPeerToIntMap = new HashMap<String, Integer>();
		MessageDistributor.createLocalPeerMap(this.localPeerMap, this.localPeerToIntMap, peer.getAllPeerNames(), peer.getPeerName());

		if (!this.isLocalMode) {
			this.peerMap = new HashMap<Integer, String>();
			this.peerToIntMap = new HashMap<String, Integer>();
			MessageDistributor.createPeerMap(this.peerMap, this.peerToIntMap, peer.getAllPeerNames(), masterPeer);
		} else {
			this.peerMap = null;
			this.peerToIntMap = null;
		}
	}

	/**
	 * Returns the total number of peers (threads) running on this node.
	 * 
	 * @return int greater zero
	 */
	protected int getNumberOfLocalPeers() {
		if (this.localPeerMap != null) {
			return this.localPeerMap.entrySet().size();
		}
		return 1;
	}

	/**
	 * Returns the total number of nodes (different PCs) used by this bsp job.
	 * 
	 * @return int greater zero
	 */
	protected int getNumberOfNodes() {
		if (this.peerMap != null) {
			return this.peerMap.entrySet().size();
		}
		return 1;
	}

	/**
	 * Returns true on local mode.
	 * 
	 * @return
	 */
	private boolean isLocalŔunningMode() {

		// final boolean localfs =
		// Training_Helper.isLocalFileSystem(this.peer.getConfiguration());

		boolean local = false;
		for (final String peer : this.peer.getAllPeerNames()) {
			if (peer.contains("local")) {
				local = true;
				break;
			}
		}

		// if (localfs || local) {
		// return true;
		// }
		if (local) {
			return true;
		}

		return false;
	}

	/**
	 * Distribute one message to all peers.
	 * 
	 * @param msg
	 *            the message to distribute.
	 * @throws IOException
	 *             during sending the messages
	 * @throws InterruptedException
	 *             during sync phase
	 * @throws SyncException
	 *             during sync phase
	 */
	public abstract HamaMessage distributeMessageToAllPeers(String senderName, HamaMessage msg, String[] peerNames) throws IOException,
			SyncException, InterruptedException;

	/**
	 * Aggregate messages from all peers and send result to master peer. When
	 * this method returns the master is having one new message containing the
	 * sum from all messages.
	 * 
	 * @param msg
	 *            the aggregated message or null
	 * @throws IOException
	 * @throws InterruptedException
	 * @throws SyncException
	 */
	public abstract HamaMessage distributeAndAggegateMessageToMasterPeer(HamaMessage msg, String masterPeer) throws IOException,
			SyncException, InterruptedException;

	/**
	 * Creates two maps containing only the peers/threads from this node.
	 * 
	 * @param peerMap
	 *            map resolves peer name to unique number.
	 * @param peerToIntMap
	 *            map resolves unique number to peer name.
	 * @param peerNames
	 *            full names of all peers
	 * @param masterPeer
	 *            full name of master peer
	 */
	protected static void createLocalPeerMap(final Map<Integer, String> peerMap, final Map<String, Integer> peerToIntMap,
			final String[] peerNames, final String localpeerName) {
		Arrays.sort(peerNames);

		int peerNumber = 1;

		final String localName = localpeerName.split(":")[0];
		Arrays.sort(peerNames);
		for (final String peerName : peerNames) {
			final String[] splitted = peerName.split(":");
			final String hostName = splitted[0];

			if (!localName.equals(hostName)) {
				continue;
			}

			peerMap.put(peerNumber, peerName);
			peerToIntMap.put(peerName, peerNumber);

			peerNumber++;
		}
	}

	/**
	 * Creates two maps containing only the first peer/thread of each node.
	 * 
	 * @param peerMap
	 *            map resolves peer name to unique number.
	 * @param peerToIntMap
	 *            map resolves unique number to peer name.
	 * @param peerNames
	 *            full names of all peers
	 * @param masterPeer
	 *            full name of master peer
	 */
	protected static void createPeerMap(final Map<Integer, String> peerMap, final Map<String, Integer> peerToIntMap,
			final String[] peerNames, final String masterPeer) {
		Arrays.sort(peerNames);

		int peerNumber = 2;
		final String masterName = masterPeer.split(":")[0];
		peerMap.put(1, masterPeer);
		peerToIntMap.put(masterPeer, 1);

		String lastPeer = "";

		Arrays.sort(peerNames);
		for (final String peerName : peerNames) {
			final String[] splitted = peerName.split(":");
			final String hostName = splitted[0];

			if (hostName.equals(masterName) || lastPeer.equals(hostName)) {
				continue;
			}

			peerMap.put(peerNumber, peerName);
			peerToIntMap.put(peerName, peerNumber);

			lastPeer = hostName;
			peerNumber++;
		}
	}

	protected BSPPeer<?, ?, ?, ?, HamaMessage> getPeer() {
		return this.peer;
	}

	protected String getMasterPeer() {
		return this.masterPeer;
	}

	protected Map<Integer, String> getPeerMap() {
		return this.peerMap;
	}

	protected Map<String, Integer> getPeerToIntMap() {
		return this.peerToIntMap;
	}

	protected Map<Integer, String> getLocalPeerMap() {
		return this.localPeerMap;
	}

	protected Map<String, Integer> getLocalPeerToIntMap() {
		return this.localPeerToIntMap;
	}

	protected boolean isLocalMode() {
		return this.isLocalMode;
	}
}
